package streaming.api.sink;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import streaming.api.beans.SensorReading;
import utils.PropertiesReader;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.util.Properties;

/**
 * kafka -> MySQL-JDBC
 * 数据来源： kafka [myTest]
 * Sink目标： mysql(jdbc) 库[zzb]表[flink_test]
 */
public class SinkTest2_jdbc_2 {

    private static String kafkaServers = PropertiesReader.get("default.kafka.servers");
    private static String topicFrom = PropertiesReader.get("default.kafka.topic.json.C");

    private static String db_url = PropertiesReader.get("target.db.mysql.url");
    private static String db_username = PropertiesReader.get("target.db.mysql.username");
    private static String db_password = PropertiesReader.get("target.db.mysql.password");

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // kafka 输入源配置
        Properties props = new Properties();
        props.setProperty("bootstrap.servers", kafkaServers);
        props.setProperty("group.id",  "flink-test-SinkTest2_jdbc_2");
        // K10001,1000,32.66
        FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>(topicFrom, new SimpleStringSchema(), props);
        DataStream<String> inputStream = env.addSource(kafkaConsumer);

        DataStream<SensorReading> dataStream = inputStream.map(line -> {
            String[] fields = line.split(",");
            return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
        });
        inputStream.print("data:");

        dataStream.addSink(new MyJdbcSink());

        env.execute();
    }

    public static class MyJdbcSink extends RichSinkFunction<SensorReading> {

        // 声明连接和预编译语句
        Connection connection = null;
        PreparedStatement insertStmt = null;
        PreparedStatement updateStmt = null;

        @Override
        public void open(Configuration parameters) throws Exception {
            connection = DriverManager.getConnection(db_url, db_username, db_password);
            insertStmt = connection.prepareStatement("insert into flink_test (id, temp) values (?, ?)");
            updateStmt = connection.prepareStatement("update flink_test set temp = ? where id = ?");
        }

        // 每来一条数据，调用连接，执行sql
        @Override
        public void invoke(SensorReading value, Context context) throws Exception {
            // 直接执行更新语句，如果没有更新那么就插入
            updateStmt.setDouble(1, value.getTemperature());
            updateStmt.setString(2, value.getId());
            updateStmt.execute();
            if (updateStmt.getUpdateCount() == 0) {
                insertStmt.setString(1, value.getId());
                insertStmt.setDouble(2, value.getTemperature());
                insertStmt.execute();
            }
        }

        @Override
        public void close() throws Exception {
            if (insertStmt != null ) {
                insertStmt.close();
            }
            if (updateStmt != null ) {
                updateStmt.close();
            }
            if (connection != null ) {
                connection.close();
            }
        }

    }

}
